package com.xiaotao.realtimeui.runner;

import com.xiaotao.realtimeui.po.PaymentInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Component
@Slf4j
public class PaymentInfoProducer implements ApplicationRunner, Runnable {
    private static final Properties props = new Properties();
    private final KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
    private final PaymentInfo paymentInfo = new PaymentInfo();
    static {
        props.put("bootstrap.servers", "master:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 5);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Thread thread = new Thread(this);
        thread.start();
    }


    public void send() {
        try {
            String msg = paymentInfo.random();
            producer.send(new ProducerRecord<>("itcast_order", msg));
            log.info("数据已发送到Kafka:" + msg);
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        while (true) {
            send();
        }
    }
}
